package com.siping.并发.黑马并发实战.自定义线程池;

import javafx.concurrent.Task;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author YangXu
 * @date 2022/2/20
 * @description: 阻塞队列，用于存放线程池的任务
 */
public class BlockingQueue<T> {
    /**
     * 队列
     */
    private Deque<T> queue = new ArrayDeque<>();
    /**
     * 锁
     */
    private ReentrantLock lock = new ReentrantLock();
    /**
     * 生产者条件变量，当任务满了之后进行等待
     */
    private Condition fullWaitSet = lock.newCondition();
    /**
     * 消费者条件变量，当任务没有了之后进行等待
     */
    private Condition emptyWaitSet = lock.newCondition();
    /**
     * 容量
     */
    private int capacity;

    public BlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    /**
     * 取出一个任务
     *
     * @param timeout
     * @param timeUnit
     * @return
     */
    public T poll(long timeout, TimeUnit timeUnit) {
        long nanos = timeUnit.toNanos(timeout);
        lock.lock();
        try {
            while (queue.isEmpty()) {
                try {
                    if (nanos <= 0) {
                        // 超时还未取到
                        return null;
                    }
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T task = queue.poll();
            fullWaitSet.signal();
            return task;
        } finally {
            lock.unlock();
        }
    }

    public void offer(T task, long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while (queue.size() == capacity) {
                // 任务队列已满，进入等待
                try {
                    if (nanos <= 0) {
                        return;
                    }
                    System.out.println("等待放入任务...");
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("任务队列添加任务...");
            queue.addLast(task);
            emptyWaitSet.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 添加任务
     *
     * @param task
     */
    public void put(T task) {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                // 任务队列已满，进入等待
                try {
                    System.out.println("等待放入任务...");
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("任务队列添加任务...");
            queue.addLast(task);
            emptyWaitSet.signal();
        } finally {
            lock.unlock();
        }
    }

    public void tryPut(T task, RejectPolicy<T> rejectPolicy) {
        lock.lock();
        try {
            if (queue.size() == capacity) {
                rejectPolicy.reject(this, task);
            } else {
                System.out.println("任务队列添加任务...");
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        } finally {
            lock.unlock();
        }
    }
}
